/**
 * Project - The Three Phase Commit Protocol
 * Authors - Namrata Baliga, Aditya Joshi
 **/
import java.io.*;
import java.util.*;
import java.util.concurrent.*;
import java.net.*;

// The Participant class encompasses the coordinator
// and participant roles.
class Participant
{
    int port,pid;
    String logFile;
    // ArrayList of processes a node believes to be operational
    ArrayList<Integer> upSet;
    //Current state of the process. Commit/Abort/Uncertain
    String currentState;
    // the state request response
    String stateReqResp;
    // socket listening to incoming connections
    ServerSocket pSocket;
    // keeping track of number of responses got
    int allResp;
    // semaphore signifies threads waiting for all responses to come to awaken 
    boolean done;
    // number of participants in the protocol
    int numParticipants; 
   
    // keeping track of coordinator
    int coordinator;

    // Creating a new node

    public Participant(int pid, int port)
    {
        this.pid = pid;
        this.port = port;
        // 0 to indicate responses not got from all
        this.allResp=0;
        // false  that we r not doen yet
        this.done = false;
        this.numParticipants = 0;
        // Log file to hold protocol state in stable storage
        // Structure: state;upset. eg: ABORT;1:2000,2:3000
        this.logFile = ""+pid+".log";
        this.upSet = new ArrayList<Integer>();
        this.currentState = Const.START;
        this.stateReqResp = Const.UNCERTAIN;
        System.out.println("PID: "+pid+" - Created new node at port: "+port);
        this.coordinator = -1;
    }

    // reading the log at recovery
    public void readLog()
    {
        //System.out.println("DEBUG: "+pid+" - Reading log information ");
        try
        {
            // Initialize upset to 0's. At time of recovery the upSet volatile
            // information is lost.
            for (int j=0;j<10;j++)
            {
                upSet.add(0);
            }
            BufferedReader br = new BufferedReader(new FileReader(logFile));
            String lastLine="",line="";
            // Go till the last "not null" line of the logfile
            do
            {
                lastLine = line;
                line = br.readLine();
            }while (line!=null);
            //System.out.println("DEBUG: "+pid+" - last log line"+lastLine);
            // The logfile splitter
            StringTokenizer logTokens = new StringTokenizer(lastLine, ";");
            // updating the state
            currentState = logTokens.nextToken();
            //System.out.println("DEBUG: "+pid+" - state from log file"+currentState);
            // getting the upset
            StringTokenizer upSetTokens =  new StringTokenizer(logTokens.nextToken(), ",");
            while (upSetTokens.hasMoreTokens())
            {
                StringTokenizer pidPort = new StringTokenizer(upSetTokens.nextToken(), ":");
                upSet.set(Integer.parseInt(pidPort.nextToken()),
                        Integer.parseInt(pidPort.nextToken()));
            }
            //System.out.println("DEBUG: "+pid+" - Done reading log information ");
        }
        catch(IOException io)
        {
            System.out.println("PID: "+pid+" - Caught io exception: "+io.getMessage());
        }

    }

    public String upSetListToString()
    {
        Integer upSetArr [] =  upSet.toArray(new Integer [upSet.size ()]);
        String upSetConcat="";
        // creating a upSet string that is comma separated
        // concatenation of all pid:port pairs
        for (int i=0;i<upSetArr.length;i++)
        {
            if (upSetArr[i]!=0)
            {
                if (upSetConcat.equals(""))
                {
                    upSetConcat+=i+":"+upSetArr[i];
                }
                else
                {
                    upSetConcat+=","+i+":"+upSetArr[i];
                }
            }    
        }
        return upSetConcat;
    }

    public void upSetStrToList(String upSetStr)
    {
       // System.out.println("got: "+upSetStr);
        StringTokenizer upSetTokens =  new StringTokenizer(upSetStr, ",");
      //  System.out.println("upset size: "+upSet.size());
        while (upSetTokens.hasMoreTokens())
        {
            StringTokenizer pidPort = new StringTokenizer(upSetTokens.nextToken(), ":");
            synchronized(this)
            {
                upSet.set(Integer.parseInt(pidPort.nextToken()),
                    Integer.parseInt(pidPort.nextToken()));
            }
        }
    }
    // Writing to the log. Synchronized to handle threads
    public synchronized void writeLog(String state,ArrayList<Integer> upSet)
    {
        try
        {
            // Writing to a log file
            BufferedWriter bw = new BufferedWriter(new FileWriter(logFile,true));
            
            // write to file
            bw.write(state+";"+upSetListToString());
            bw.newLine();
            bw.flush();
            bw.close();
            System.out.println("PID: "+pid+" - Wrote to log: "+state+";"+upSetListToString());
        }
        catch(IOException io)
        {
            System.out.println("PID: "+pid+" - Caught io: "+io.getMessage());
        }
    }

    public synchronized void assignStateReqStatus()
    {
        //System.out.println("curr: "+currentState);
        if (currentState.equals(Const.START)
                || currentState.equals(Const.VOTE_NO)
                || currentState.equals(Const.ABORT))
        {
            stateReqResp = Const.ABORTED;
        }
        if (currentState.equals(Const.COMMIT))
        {
            
            stateReqResp = Const.COMMITTED; 
        }
        else if(currentState.equals(Const.PRE_COMMIT))
        {
            stateReqResp = Const.COMMITABLE;
        }
        else
        {
            stateReqResp = Const.UNCERTAIN;
        }
        //System.out.println("state  req assigned: "+stateReqResp);
    }

    public void invokeParticipant(String stage, ExecutorService threadExecutor) 
        throws SocketTimeoutException, IOException, InterruptedException, ExecutionException
    {
        ThreadParticipant server = new ThreadParticipant(this,stage);
        new Thread(server).start();
        BufferedWriter bw = new BufferedWriter(new FileWriter(this.pid+Const.DEC_FILE));
        BufferedReader br = new BufferedReader(new FileReader(this.pid+Const.DEC_FILE));
        String read="";
        try 
        {
            bw.write("");
            bw.flush();
            while (read == null || read.equals(""))
            {
                read = br.readLine();
                //System.out.println(" read: "+read);
                Thread.sleep(1 * 1000);
            };
            bw.write("");
            bw.flush();
            bw.close();

        } catch (Exception e) {
            e.printStackTrace();
        }
        //System.out.println("Stopping Server");
        server.stop();
        if (read.equals(Const.REQ_TIMEOUT))
        {
            throw new SocketTimeoutException();
        }
        else
        {
            //System.out.println("got a ur so throwing exception");
            throw new ExecutionException(Const.UR_ELECTED,null);
        }
    }

    public int initiateElection()
    {
        int i=0, min_pid=pid, decision=0;
        while(i<upSet.size())
        {
            if(upSet.get(i)!=Const.DEAD_HOST && i<min_pid)
            {
                min_pid=i;
            }
            i++;
        }
        int j=0;
        while(j<min_pid)
        {
            upSet.set(j, 0);
            j++;
        }
       if(min_pid==pid)
        {
            System.out.println("PID: "+pid+" - I am new co-ordinator");
            decision = Const.IM_COORD;
            coordinator = pid;
            //initiate termination protocol as co-ordinator
                       try
                     {
                         //System.out.println("sleeping to let others become parti");
                         Thread.sleep(3000);
                     }
                     catch(Exception e)
                     {
                     }
      }
        else
        {
            if (upSet.get(min_pid) != Const.DEAD_HOST)
            {
                System.out.println("PID: "+pid+" - Process "+min_pid+" is the new co-ordinator");
                PrintWriter out = null;
                try
                {
                    Socket pSocket = new Socket("localhost",upSet.get(min_pid));
                    out = new PrintWriter(pSocket.getOutputStream(), true);
                    out.println(Const.UR_ELECTED);
                    System.out.println("PID: "+pid+" - Sent "+Const.UR_ELECTED+" to process "+min_pid);
                }
                catch (Exception e)
                {
                    System.out.println("PID: "+pid+" - Sent "+Const.UR_ELECTED);
                }
               //participate in termination as participant
                decision = Const.IM_PARTICIPANT;
                System.out.println("PID: "+pid+" - I am now a participant ");
                coordinator = min_pid;
                //try{
                //Thread.sleep(5000);
                //}
                //catch(Exception e)
                //{
                //}

            }
        }
        //remove all lower processes from upSet
        return decision;
    }


    public void invokeCoordinator(String stage, ExecutorService threadExecutor)
    {
        // Coordinator action
        // Proactively connect to all participants
        //System.out.println("PID: "+pid+" - Started coordinator mode");
        //writeLog(stage,upSet);
        Set<Future<Integer>> set = new HashSet<Future<Integer>>();
        for (int i = 0;i<upSet.size();i++)
        {
         //   System.out.println("Invoked for node: "+i+" upSet:"+upSet.get(i));
            if (upSet.get(i)!=0 && i !=pid)
            {
                numParticipants++;
                FutureTask<Integer> task = new FutureTask<Integer> 
                    (new ThreadCoordinator(this,i,upSet.get(i),stage));
           //     System.out.println("Invoked for node: "+i);
                set.add(task);
                threadExecutor.submit(task);
            }   
        }
        if (numParticipants == 0)
        {
            System.out.println("PID: "+pid+" - No participant alive. Basing decision on my state.");
            assignStateReqStatus();
            if (stateReqResp.equals(Const.COMMITABLE))
            {
                System.out.println("PID: "+pid+" - Writing COMMIT to log");
                writeLog(Const.COMMIT,upSet);
            }
            else if (stateReqResp.equals(Const.COMMITTED))
            {
                System.out.println("PID: "+pid+" - Writing COMMIT to log");
                writeLog(Const.COMMIT,upSet);
            }
            else if (stateReqResp.equals(Const.ABORTED))
            {
                System.out.println("PID: "+pid+" - Writing ABORT to log");
                writeLog(Const.ABORT,upSet);
            }
            else if (stateReqResp.equals(Const.UNCERTAIN))
            {
                System.out.println("PID: "+pid+" - Writing ABORT to log");
                writeLog(Const.ABORT,upSet);
            }

        }
        int retVal =0;
        try
        {
             for (Future<Integer> task : set)
             {
                 retVal = task.get();
             }
        }
        catch(InterruptedException io)
        {
            System.out.println("Caught: "+io.getMessage());
            //io.printStackTrace();
        }
        catch(ExecutionException io)
        {
            System.out.println("Caught:"+io.getMessage());
            //io.printStackTrace();
        }   
    }

    public int recover()
    {
       System.out.println("PID: "+pid+" - I am recovering from crash ");
        // recover from log file
        readLog();
        //System.out.println("read state: "+currentState);
    	if(currentState.equals(Const.VOTE_REQ)
                ||currentState.equals(Const.ABORT)||currentState.equals(Const.ABORT))
        {
          //  System.out.println("inside vr"); 
            //failed before sending yes/no to co-ordinator or it failed after receiving abort. Hence unilaterally abort
            System.out.println("PID: "+pid+" - Reading log to recover. Safe to ABORT.");
            writeLog(Const.ABORT,upSet);
            //initiate abort sequence
            return -1;
        }    
        else if(currentState.equals(Const.COMMIT))
        {
             System.out.println("PID: "+pid+" - Reading log to recover. Safe to COMMIT.");
           writeLog(Const.COMMIT,upSet);
            return -1;
        }
        else if(currentState.equals(Const.VOTE_YES) || currentState.equals(Const.PRE_COMMIT))
        {
            //cannot decide on its own. need to contact other processes
            //String upSetstr=upSetListToString();
            boolean totalFailure=true;
            Socket mSocket=null;
            PrintWriter out=null;
            BufferedReader in=null;
            String[] decResponse=new String[2];
            System.out.println("PID: "+pid+" - Cannot recover independently. Contacting other participants...");
            try
            {
                Thread.sleep(5000);
            }
            catch(Exception e)
            {
            }
            while(true)	
            {
                for(int i=0;i<upSet.size();i++)
                {
                    if(upSet.get(i)!=0)
                    {
                        try{
                        mSocket=new Socket("localhost",upSet.get(i));
                           } 	
                        catch (IOException e)
                            {
                          //      System.err.println("Couldn't get io connection");
                    //            System.exit(1);
                            continue;
                            }
                        try{	
                            out = new PrintWriter(mSocket.getOutputStream(), true);
                            in = new BufferedReader(new InputStreamReader(mSocket.getInputStream()));
                        }catch(IOException io)
                        {
                        }
                        /*try{
                        mSocket.setSoTimeout(10000);
                        }catch(IOException ste)
                        {
                        }*/
                        System.out.println("PID: "+pid+" - Querying participant :"+i+" for decision");
                        out.println(Const.DECISION_REQ);
                            // read response
                            String readData=null;
                            try
                        {
                                readData = in.readLine();
                            decResponse=decReqParse(readData);
                            readData=decResponse[0];
                            totalFailure=false;
                        }
                        catch(SocketTimeoutException ste)
                        {
                          //  System.out.println("no response from process " +i);
                            continue;
                        }
                        catch(IOException io)
                        {
                        }
                        System.out.println("PID: "+pid+" - Obtained from participant: "+i+" - "+readData);
                        if(readData.equals(Const.ABORT))
                        {
                            //other process has decided abort. I too can abort
                            System.out.println("PID: "+pid+" - Safe to ABORT");
                            writeLog(Const.ABORT,upSet);
                            //initiate abort sequence
                            return -1;
                        }
                        else if(readData.equals(Const.COMMIT))
                        {
                             System.out.println("PID: "+pid+" - Safe to COMMIT");
                           writeLog(Const.COMMIT,upSet);	
                            return -1;
                        }
                         
                    }	
                    
                }//for loop
                if(totalFailure==true)
                    break;// now have to detect last process to fail
                else
                {	//wait before polling everyone again
                    long t0,t1;
                        t0=System.currentTimeMillis();
                     System.out.println("PID: "+pid+" - No decision obtained yet. Trying again.");
                       do
                    {
                            t1=System.currentTimeMillis();
                        }
                        while (t1-t0<10000);
                }	
            }
        // to find last process to fail
            ArrayList<Integer> intersect=new ArrayList<Integer>();
            ArrayList<Integer> currentAlive=new ArrayList<Integer>();
            ArrayList<Integer> finalIntersect=new ArrayList<Integer>();
            for (int m=0;m<Const.MAX_PARTICIPANTS;m++)
            {
                intersect.add(0);
                currentAlive.add(0);
                finalIntersect.add(0);
            }
    //		bool lastProcAlive=true;
            currentAlive.set(this.pid,this.port);
label1:
            for(int i=0;i<upSet.size();i++)
            {	Socket mySocket=null;
                PrintWriter out1=null;
                BufferedReader in1=null;
    //			static int index=0;
                if(upSet.get(i)!=0)
                {
                    try{
                    mySocket=new Socket("localhost",upSet.get(i));
                    }
                    catch(IOException io)
                    {
                        //System.err.println("Could not connect to process" +i);
                    continue;
                    }
                    try{
                    out1 = new PrintWriter(mySocket.getOutputStream(), true);
                    in1 = new BufferedReader(new InputStreamReader(mySocket.getInputStream()));
                    }catch(IOException io)
                    {
                    }
                //	pSocket.setSoTimeout(10000);
                    out1.println(upSet);
                    String inUpSetstr=new String();
                    try
                    {	 // read response in separate thread
                             inUpSetstr = in1.readLine();
                         decResponse=decReqParse(inUpSetstr);
                         inUpSetstr=decResponse[1];
                         currentAlive.set(i,upSet.get(i));
                    }
                    catch(SocketTimeoutException ste)
                    {
                       // System.out.println("no response from process " +(i-1));
                        continue;
                    }
                    catch(IOException io)
                    {
                    }
                        //System.out.println(" Obtained from process:"+i);
                        ArrayList<Integer> inUpSet=upSetStrToListRet(inUpSetstr);
                    for(int k=0;k<inUpSet.size();k++)
                        System.out.println(inUpSet.get(i));
                    if(i==0)
                        findIntersect(upSet, inUpSet, intersect);
                    else    
                        findIntersect(intersect,inUpSet,finalIntersect);
                    
                }//if
                if(i==upSet.size()-1)
                {
                //    System.out.println("abcd");
                    // checking if no process alive yet!!
                  //  System.out.println("aliveness");
                    int noAlive = 0;
                    for (int l=0;l<Const.MAX_PARTICIPANTS && (noAlive == 0);l++)
                    {
                  //      System.out.println(currentAlive.get(l));
                        if (currentAlive.get(l) != 0 && pid !=l)
                        {
                    //        System.out.println("am alive: "+l);
                            noAlive = 1;
                        }
                    }
                    if (noAlive == 0)
                    
                    { //  System.out.println("no alive");
                        i = 0;
                        return -1;
                        //continue; 
                    }
                    
                    //check if set of currently alive processes include the intersect
                    int min_size=finalIntersect.size()<currentAlive.size()?finalIntersect.size():currentAlive.size();
                //	lastProcAlive=true;
            
                    for(int h=0;h<min_size;h++)
                    if(((currentAlive.get(h)==0) && (finalIntersect.get(h)!=0)))//if there is a case where an intersecting process is not alive;which means the last process(es) to fail is still not up
                    {
                //		lastProcAlive=false;
                        //wait for sometime and again check on this process
                        //System.out.println("time loop");
                        long t0,t1;
                            t0=System.currentTimeMillis();
                            do
                        {
                                    t1=System.currentTimeMillis();
                            }
                            while (t1-t0<5000);
                        i=h-1;
                        break;
                    }
                }
            }//for		
                //all in the intersection set are up	
            Socket vSocket=null;
            PrintWriter out2=null;
            BufferedReader in2=null;

            for(int v=0;v<finalIntersect.size();v++)
            {//contact every process in the intersect set to find out if it knows the decision
                 try{
                vSocket=new Socket("localhost",finalIntersect.get(v));
                }
                catch(IOException io)
                {
                   // System.err.println("Could not connect to process" +v);
                continue;
                }
                try{
                out2 = new PrintWriter(vSocket.getOutputStream(), true);
                in2 = new BufferedReader(new InputStreamReader(vSocket.getInputStream()));
                }catch(IOException io)
                {
                }
            //	pSocket.setSoTimeout(10000);
                out2.println(Const.DECISION_REQ);
                String readDec=new String();
                try
                {	 // read response in separate thread
                         readDec = in2.readLine();
                }
                catch(SocketTimeoutException ste)
                {
                    System.out.println("no response from process " +(v-1));
                    continue;
                }
                catch(IOException io)
                {
                }
                System.out.println("PID: "+pid+" - Obtained from process "+v+":"+readDec);
                if(readDec.equals(Const.ABORT))
                {
                    //other process has decided abort. I too can abort
                    writeLog(Const.ABORT,upSet);
                    //initiate abort sequence
                    return -1;
                }
                else if(readDec.equals(Const.COMMIT))
                {
                    writeLog(Const.COMMIT,upSet);	
                    return -1;
                }
                    
            }//for
        //not got decision from last processto fail. hence start termination here (election protocol)
        return this.initiateElection();
        //call co-ord or participant's terminaton algo ???
        }
        return -1;
        
    }

	public void findIntersect(ArrayList<Integer> upSet1,ArrayList<Integer> upSet2,ArrayList<Integer>targetUpSet)
	{
		int index=0;
		int min_size=(upSet1.size()<upSet2.size()?upSet1.size():upSet2.size());
		for(int k=0;k<min_size;k++)
		{
			if((upSet1.get(k)!=0) && (upSet2.get(k)!=0))	
				targetUpSet.set(k,upSet1.get(k));
		}
	}

	 public ArrayList<Integer> upSetStrToListRet(String upSetStr)
    {
       // System.out.println("got: "+upSetStr);
	ArrayList<Integer> mUpSet = new ArrayList<Integer>();
        StringTokenizer upSetTokens =  new StringTokenizer(upSetStr, ",");
 //       System.out.println("upset size: "+upSet.size());
        while (upSetTokens.hasMoreTokens())
        {
            StringTokenizer pidPort = new StringTokenizer(upSetTokens.nextToken(), ":");
            synchronized(this)
            {
                mUpSet.set(Integer.parseInt(pidPort.nextToken()),
                    Integer.parseInt(pidPort.nextToken()));
            }
        }
	return mUpSet;
    }

	 public String []  decReqParse(String decReq)
    {
//	System.out.println("got: "+decReq);
	String[] decReqParsed=new String[2];
        StringTokenizer decReqTokens =  new StringTokenizer(decReq, ";");
 //       System.out.println("upset size: "+upSet.size());
        while (decReqTokens.hasMoreTokens())
        {   String temp= decReqTokens.nextToken();
            //StringTokenizer pidPo = new StringTokenizer(upSetTokens.nextToken(), ":");
            synchronized(this)
            {
               decReqParsed[0]=decReqTokens.nextToken();
               decReqParsed[1]=decReqTokens.nextToken();
            }
        }
	return decReqParsed;
    }

    public void cleanUp()
    {
        System.out.println("PID: "+pid+" - Cleaning up process information");
        File f = new File(logFile);
        if (f.exists())
        {
            f.delete();
        }
        System.out.println("PID: "+pid+" - Clean up complete");
    }
}

class WorkerRunnable implements Runnable
{
    protected Socket cSocket = null;
    protected String stage   = null;
    protected Participant node = null;

    public WorkerRunnable(Participant node, Socket clientSocket, String stage) {
        this.cSocket = clientSocket;
        this.node   = node;
        this.stage = stage;
    }
    
    public String stateRequestAlgo(String whichSR, BufferedReader in, PrintWriter out) 
    {
        int currCoord = node.coordinator;
        synchronized(this)
        {
            //node.currentState = Const.STATE_REQ;
        }
        try
        {
            String readData = "";
  //          System.out.println("inside: "+whichSR);
            if (whichSR.equals(Const.STATE_REQ))
            {
                System.out.println("PID: "+node.pid+" - Waiting for "+Const.STATE_REQ);
                // waiting for state request
                if ((readData = in.readLine()) == null)
                {
                    //throw new SocketTimeoutException();
                    if (currCoord == node.coordinator)
                        return Const.REQ_TIMEOUT;
                }
                System.out.println("PID: "+node.pid+" - Obtained "+readData+" from coordinator");
                StringTokenizer stateReqToken = new StringTokenizer(readData, ";");
                stateReqToken.nextToken(); stateReqToken.nextToken(); 
                synchronized(this)
                {
                    node.coordinator = Integer.parseInt(stateReqToken.nextToken());
                    StringTokenizer upSetTokens = new StringTokenizer(readData,";");
                    upSetTokens.nextToken(); 
                    node.upSetStrToList(upSetTokens.nextToken());
                    //node.writeLog(Const.STATE_REQ,node.upSet);               
                }
            }
    //        System.out.println("before assign");
            node.assignStateReqStatus();
            System.out.println("PID: "+node.pid+" - Sending state "+node.stateReqResp+" to coordinator."
                    +"\n Waiting for decision from coordinator");
            out.println(node.stateReqResp);
            if ((readData = in.readLine()) == null)
            {
                 if (currCoord == node.coordinator)
                    return Const.REQ_TIMEOUT;
           }
            System.out.println("PID: "+node.pid+" - Obtained "+readData+" from coordinator");
            if (readData.equals(Const.ABORT))
            {
                node.writeLog(Const.ABORT,node.upSet);
                synchronized (this)
                {
                    node.currentState = Const.ABORT;
                }

            }
            else if (readData.equals(Const.COMMIT))
            {
                synchronized (this)
                {
                    node.currentState = Const.COMMIT;
                }
                node.writeLog(Const.COMMIT,node.upSet);
            }
            else
            {
                System.out.println("PID: "+node.pid+" - Obtained "+readData+" from coordinator");
                out.println(Const.ACK);
                System.out.println("PID: "+node.pid+" - Sending "+Const.ACK+" to coordinator. \n Waiting for: "
                        +Const.COMMIT);
                if((readData = in.readLine()) == null)
                {
                    if (currCoord == node.coordinator)
                        return Const.REQ_TIMEOUT;
                }
                System.out.println("PID: "+node.pid+" - Obtained "+readData+" from coordinator");
                node.writeLog(Const.COMMIT,node.upSet);
            }
        }
        catch(Exception e)
        {
            System.out.println("PID: "+node.pid+" - Caught error "+e.getMessage());
        }
        return Const.NOOP;
    }

    public synchronized void writePartiDecision(String decision)
    {
         try
         {
             BufferedWriter bw = new BufferedWriter(new FileWriter(node.pid+Const.DEC_FILE));
             bw.write(decision);
             bw.flush();
             bw.close();
         }
         catch (IOException e)
         {
         }
    }

    public void run() 
    {
        try 
        {
            // Participant waits for coordinator
        //    System.out.println("PID: "+node.pid+" - New incoming connection");
            String readData = "";
            // Set i/o streams
            PrintWriter out = new PrintWriter(cSocket.getOutputStream(), true);
            BufferedReader in = new BufferedReader(new InputStreamReader(
                        cSocket.getInputStream()));
            // read response
            cSocket.setSoTimeout(Const.TIMEOUT); 
            
            // read what got
            // if state request do state request algo
            if ((readData = in.readLine()) == null)
            {
          //      System.out.println("here");
                synchronized(this)
                {
            //        System.out.println("curr coord: "+node.coordinator+" of :"+node.pid);
                    if (node.coordinator != -1)
                    {
                        node.upSet.set(node.coordinator, Const.DEAD_HOST);
                    }
                }
                //throw new SocketTimeoutException();
                writePartiDecision(Const.REQ_TIMEOUT);
                return;
            }
            System.out.println("PID: "+node.pid+" - Obtained "+readData);
            if (readData.indexOf(Const.VOTE_REQ) != -1)
            {
                StringTokenizer upSetTokens = new StringTokenizer(readData,";");
                upSetTokens.nextToken(); 
                node.upSetStrToList(upSetTokens.nextToken());
                int currCoord = 0;
                synchronized (this)
                {
                    node.coordinator = Integer.parseInt(upSetTokens.nextToken());
                    currCoord = node.coordinator;
                }

                node.writeLog(Const.VOTE_REQ,node.upSet);               
                Scanner stdin = new Scanner(System.in);
                System.out.print("PID: "+node.pid+" - Enter Vote (yes/no): ");
                String vote = stdin.nextLine();
                System.out.print("PID: "+node.pid+" - Sent Vote "+vote+" to coordinator");
                out.println(vote.toUpperCase());             
                synchronized (this)
                {
                    node.writeLog(vote.toUpperCase(),node.upSet);               
                    node.currentState = vote.toUpperCase();
                }
                if (vote.compareToIgnoreCase(Const.VOTE_YES) == 0)
                {
                    System.out.println("PID: "+node.pid+" - Waiting for ABORT or PRECOMMIT ");
                    if ((readData = in.readLine()) == null)
                    {
           //             System.out.println("bf excep: curr:"+currCoord+" new: "+node.coordinator);
                        if (node.coordinator > currCoord)
                        {
                            System.out.println("Pid: "+node.pid+" Detected another instance of "+
                                   " transaction. CLOSING CURRENT THREAD.");
                        }
                        else
                        {
             //               System.out.println("writing some");
                            writePartiDecision(Const.REQ_TIMEOUT);
                        }
                        return;
                       // throw new SocketTimeoutException();
                    }
                    System.out.println("PID: "+node.pid+" - Obtained "+readData+" from coordinator");
                    if (readData.equals(Const.ABORT))
                    {
                        System.out.println("PID: "+node.pid+" - Aborting transaction");
                        synchronized (this)
                        {
                            node.currentState = Const.ABORT;
                        }
                        node.writeLog(Const.ABORT,node.upSet); 
                    }
                    else
                    {
                        System.out.println("PID: "+node.pid+" - Obtained "+readData+" from coordinator");
                        System.out.println("PID: "+node.pid+" - Pre-Commiting transaction");
                         synchronized (this)
                        {
                            node.currentState = Const.PRE_COMMIT;
                        }
                       node.writeLog(Const.PRE_COMMIT,node.upSet);
                         System.out.println("PID: "+node.pid+" - Sending ACK");
                        try
                        {
                            Thread.sleep(5000);
                        }
                        catch(Exception e)
                        {
                        }
                           System.out.println("PID: "+node.pid+" - Sent ACK to coordinator");
                        out.println(Const.ACK);
                        synchronized (this)
                        {
                            //node.currentState = Const.ACK;
                        }
                        //node.writeLog(Const.ACK,node.upSet);
                        if ((readData = in.readLine()) == null)
                        {
                            writePartiDecision(Const.REQ_TIMEOUT);                       
                            return;
                            //    throw new SocketTimeoutException();
                        }
                        System.out.println("PID: "+node.pid+" - Obtained "+readData+" from coordinator");
                        System.out.println("PID: "+node.pid+" - Commiting transaction");
                        synchronized (this)
                        {
                            node.currentState = Const.COMMIT;
                        }
                       node.writeLog(Const.COMMIT,node.upSet);
                    }
                }
                else
                {
                    synchronized(this)
                    {
                        node.currentState = Const.VOTE_NO;
                    }
                    node.writeLog(Const.VOTE_NO, node.upSet);
                    System.out.println("PID: "+node.pid+" - ABORTing transaction");
                    synchronized(this)
                    {
                        node.currentState = Const.ABORT;
                    }
                    node.writeLog(Const.ABORT,node.upSet);
                }
                System.out.println("PID: "+node.pid+" - 3PC Transaction Completed");
            }
            else if (readData.equals(Const.UR_ELECTED))
            {
                //throw new ExecutionException(Const.UR_ELECTED,null);
                writePartiDecision(Const.UR_ELECTED);
                return;
            }
            else if (readData.indexOf(Const.STATE_REQ) != -1)
            {
                //System.out.println("unsolicited SR");
                StringTokenizer stateReqToken = new StringTokenizer(readData, ";");
                stateReqToken.nextToken();  
                synchronized(this)
                {
                    node.upSetStrToList(stateReqToken.nextToken());
                    //node.writeLog(Const.STATE_REQ,node.upSet);               
                    int newCoord = Integer.parseInt(stateReqToken.nextToken());
//                    System.out.println("new: "+newCoord+" curr: "+node.coordinator);
                    if (newCoord < node.coordinator)
                    {
                        System.out.println("PID: "+node.pid+" - Encountered request "+
                                "from spurious (old) coordinator."
                                +"\n Ignoring request.");
                        return;
                    }
                    node.coordinator = newCoord; 
                    System.out.println("PID: "+node.pid+" - Initiating TERMINATION PROTOCOL with "+
                            "new coordinator: "+newCoord); 
                }
                String resp = stateRequestAlgo(Const.STATE_REQ_STARTED,in,out);
                if (!resp.equals(Const.NOOP))
                {                
                    writePartiDecision(resp);
                }
                return;
            }
            else if(readData.indexOf(Const.DECISION_REQ) != -1)
            {
                System.out.println("PID: "+node.pid+" - Responding with my current state");
                try
                {
                    Thread.sleep(5000);
                }
                catch(Exception e)
                {
                }               
                out.println(Const.DECISION_RES+";"+
                        node.currentState+";"+node.upSetListToString());
            }
            else if (stage.indexOf(Const.STATE_REQ) != -1)
            {

                String resp = stateRequestAlgo(Const.STATE_REQ,in,out);
                if (!resp.equals(Const.NOOP))
                {                
                    writePartiDecision(resp);
                }
                return;
            }
            else if (stage.indexOf(Const.RECOVER) != 1)
            {
                //node.recover();
            }
            out.close();
            in.close();
        } 
        catch (IOException e) 
        {
            //report exception somewhere.
            //e.printStackTrace();
        }
    }
}



class ThreadParticipant implements Runnable
{
    protected int          serverPort   = 8080;
    protected ServerSocket serverSocket = null;
    protected boolean      isStopped    = false;
    protected Thread       runningThread= null;
    protected Participant node = null;
    protected String stage = null;

    public ThreadParticipant(Participant node,String stage){
        this.serverPort = node.port;
        this.node = node;
        this.stage = stage;
    }

    public void run(){
        synchronized(this){
            this.runningThread = Thread.currentThread();
        }
        openServerSocket();
        while(! isStopped()){
            Socket clientSocket = null;
            try {
                clientSocket = this.serverSocket.accept();
            } catch (IOException e) {
                if(isStopped()) {
    //                System.out.println("PID: "+this.node.pid+" done") ;
                    return;
                }
                throw new RuntimeException(
                    "Error accepting client connection", e);
            }
            new Thread(
                new WorkerRunnable(
                    node,clientSocket, stage)
            ).start();
        }
      //  System.out.println("PID: "+this.node.pid+ " done.") ;
    }


    private synchronized boolean isStopped() {
        return this.isStopped;
    }

    public synchronized void stop(){
        this.isStopped = true;
        try {
            this.serverSocket.close();
        } catch (IOException e) {
            throw new RuntimeException("Error closing server", e);
        }
    }

    private void openServerSocket() {
        try {
            this.serverSocket = new ServerSocket(this.serverPort);
        } catch (IOException e) {
            throw new RuntimeException("Cannot open port : "+this.serverPort, e);
        }
    }

}

    
 
class ThreadCoordinator implements Callable 
{
    public int pid;
    public int port;
    public Participant node;
    public String stage;
    public ThreadCoordinator(Participant node, int pid, int port, String stage)
    {
        this.pid = pid;
        this.port = port;
        this.node = node;
        this.stage = stage;
    }
    
    public void incrementcurrNumResp(String resp)
    {
        synchronized (this.node)
        {
            this.node.allResp++;
            if (resp.equals(Const.VOTE_NO) || resp.equals(Const.REQ_TIMEOUT))
            {
                this.node.currentState = resp;
            }
            if (this.node.allResp == this.node.numParticipants)
            {
                this.node.done = true;
                this.node.allResp = 0;
                this.node.notifyAll();
            }
        }
    }

    public void incrementStateResp(String resp)
    {
        synchronized(this.node)
        {
            this.node.allResp++;
        //    System.out.println("got: "+resp+" i am : "+this.node.currentState);
            if (resp.equals(Const.ABORTED) || resp.equals(Const.COMMITTED))
            {
                this.node.currentState = resp;
            }
            if (resp.equals(Const.COMMITABLE) && 
                    this.node.currentState.equals(Const.UNCERTAIN))
            {
                this.node.currentState = resp;
            }
            if (this.node.allResp == this.node.numParticipants)
            {
                this.node.done = true;
                this.node.allResp = 0;
                this.node.notifyAll();
            }
        }
    }

    public Integer call() throws Exception
    {
        String sent="",recv="";
//        System.out.println("PID: "+this.node.pid+" - Connecting to process: "+this.pid);
        // connecting to participant
        Socket cSocket = new Socket("localhost",this.port);
        System.out.println("PID: "+this.node.pid+" - Process: "+this.pid+ " joins as participant");
        cSocket.setSoTimeout(Const.TIMEOUT);
        //setting in/out buffers
       PrintWriter out = new PrintWriter(cSocket.getOutputStream(), true);
        BufferedReader in = new BufferedReader(new InputStreamReader(
                                           cSocket.getInputStream()));
        
        if (this.stage.equals(Const.VOTE_REQ))
        {
            System.out.println("PID: "+this.node.pid+" - Sending vote request to participant:"+this.pid);
                    try
                {
                    Thread.sleep(5000);
                }
                catch(Exception e)
                {
                }               
           out.println(Const.VOTE_REQ+";"+this.node.upSetListToString()+";"+this.node.pid);
            // waiting for VOTE_REQ response
            try
            {
                //recv = in.readLine();
//                System.out.println("recv: "+recv);
                if ((recv = in.readLine()) == null)
                {
                    throw new SocketTimeoutException();
                }
                System.out.println("PID:"+this.node.pid+" - Participant "+this.pid+
                        " responds with "+recv);
            }
            catch(SocketTimeoutException st)
            {
                System.out.println("PID: "+this.node.pid+" - Time out from participant: "+this.pid);
                recv = Const.REQ_TIMEOUT; 
            }
            if (recv == null)
            {
                System.out.println("PID: "+this.node.pid+" - Failure of participant: "+this.pid);
                recv = Const.REQ_TIMEOUT;
            }
            incrementcurrNumResp(recv);
            synchronized (this.node)
            {
                while (!this.node.done)
                {   
                    try
                    {
                        this.node.wait();
                    }
                    catch(InterruptedException ie)
                    {
                    }
                }
            }
//            System.out.println("state:"+this.node.currentState);
            if (this.node.currentState.indexOf(Const.REQ_TIMEOUT) != -1 || 
                 this.node.currentState.indexOf(Const.VOTE_NO) != -1  )
            {
                // somebody sent a no or timedout
                System.out.println("PID: "+this.node.pid+" Got "+this.node.currentState+" so aborting run."
                        +"Sending ABORT to participant : "+this.pid);
                this.node.writeLog("ABORT",null);
                out.println(Const.ABORT);
            }
            else
            {
                // all nodes sent a yes

                System.out.println("PID: "+this.node.pid+" - Sending PRECOMMIT");
                System.out.println("PID: "+this.node.pid+" - Waiting for ACK");
                  try
                {
                    Thread.sleep(5000);
                }
                catch(Exception e)
                {
                }               
                synchronized (this)
                {
                    this.node.currentState = Const.PRE_COMMIT;
                }

                
                this.node.writeLog(Const.PRE_COMMIT,this.node.upSet);
               out.println(Const.PRE_COMMIT);
                try
                {
                    if((recv = in.readLine()) == null)
                    {
                        throw new SocketTimeoutException();
                    }
                    System.out.println("PID:"+this.node.pid+" - Got response "+recv 
                    +" from participant: "+this.pid);
                    System.out.println("PID:"+this.node.pid+" - Sending "+Const.COMMIT+" response "  
                    +" to participant: "+this.pid);
                      try
                    {
                        Thread.sleep(5000);
                    }
                    catch(Exception e)
                    {
                    }
                  this.node.writeLog(Const.COMMIT, this.node.upSet);
                    synchronized(this)
                    {
                        this.node.currentState = Const.COMMIT;
                    }
                    out.println(Const.COMMIT);
                }
                catch (SocketTimeoutException st)
                {
                    System.out.println("PID: "+this.node.pid+" - Timedout from process: "+this.pid+
                            " Ignoring process failure");
                    synchronized (this)
                    {
                        this.node.upSet.set(this.pid, 0);
                        this.node.writeLog(Const.PRE_COMMIT,this.node.upSet);
                    }
                }
            }
        }
        else
        {
             System.out.println("PID: "+this.node.pid+" - Sending "+Const.STATE_REQ+" to: "+this.pid);
        
                  try
                {
                    Thread.sleep(5000);
                }
                catch(Exception e)
                {
                }               

            // state request
            out.println(Const.STATE_REQ+";"+this.node.upSetListToString()+";"+this.node.pid);
            System.out.println("PID: "+this.node.pid+" - Waiting for STATE REPORT from: "+this.pid);
            // assign coordinator's state
            this.node.assignStateReqStatus();
            // waiting for STATE_REQ response
           // System.out.println("c state: "+this.node.currentState);
            synchronized (this)
            {
                this.node.done = false;
            }
            String readData = null;
            try
            {
                if ((readData = in.readLine()) == null)
                {
                    readData = Const.UNCERTAIN; 
                    synchronized (this)
                    {
                        this.node.upSet.set(this.pid, 0);
                        //this.node.writeLog(Const.PRE_COMMIT,this.node.upSet);
                    }
                    //throw new SocketTimeoutException();
                }               
                incrementStateResp(readData);
                synchronized (this.node)
                {

                    while (!this.node.done)
                    {   
                        try
                        {
                            this.node.wait();
                        }
                        catch(InterruptedException ie)
                        {
                        }
                    }
                }               
                if (this.node.stateReqResp.equals(Const.ABORTED) )
                {
                    this.node.writeLog(Const.ABORT,this.node.upSet);
                    System.out.println("PID: "+this.node.pid+" - Sending "+Const.ABORT+" to participant: "+this.pid);
                   try
                {
                    Thread.sleep(5000);
                }
                catch(Exception e)
                {
                }               
                   out.println(Const.ABORT);
                }
                else if (this.node.stateReqResp.equals(Const.COMMITTED) )
                {
                    this.node.writeLog(Const.COMMIT,this.node.upSet);
                    System.out.println("PID: "+this.node.pid+" - Sending "+Const.COMMIT+" to participant: "+this.pid);
                   try
                {
                    Thread.sleep(5000);
                }
                catch(Exception e)
                {
                }               
                   out.println(Const.COMMIT);
                }
                else if (this.node.stateReqResp.equals(Const.UNCERTAIN) )
                {
                    this.node.writeLog(Const.ABORT,null);
                    System.out.println("PID: "+this.node.pid+" - Sending "+Const.ABORT+" to participant: "+this.pid);
                   try
                {
                    Thread.sleep(5000);
                }
                catch(Exception e)
                {
                }               
                   out.println(Const.ABORT);
                }
                else if (this.node.stateReqResp.equals(Const.COMMITABLE) )
                {
                    System.out.println("PID: "+this.node.pid+" - Sending PRECOMMIT to participant : "+this.pid);
                    System.out.println("PID: "+this.node.pid+" - Waiting for ACK from participant : "+this.pid);
                    this.node.writeLog(Const.PRE_COMMIT,this.node.upSet);
                    try
                {
                    Thread.sleep(5000);
                }
                catch(Exception e)
                {
                }               
                  out.println(Const.PRE_COMMIT);
                    try
                    {
                        if((recv = in.readLine()) == null)
                        {
                            throw new SocketTimeoutException();
                        }
                        System.out.println("PID:"+this.node.pid+" - Obtained response "+recv 
                        +" from participant: "+this.pid);
                        System.out.println("PID:"+this.node.pid+" - Sending"+Const.COMMIT+" response "  
                        +" to participant: "+this.pid);
                        this.node.writeLog(Const.COMMIT, this.node.upSet);
                           try
                        {
                            Thread.sleep(5000);
                        }
                        catch(Exception e)
                        {
                        }               
                       out.println(Const.COMMIT);
                    }
                    catch (SocketTimeoutException st)
                    {
                        System.out.println("PID: "+this.node.pid+" - Timedout from process: "+this.pid+
                                " Ignoring process failure");
                        synchronized (this)
                        {
                            this.node.upSet.set(this.pid, 0);
                            this.node.writeLog(Const.COMMIT,this.node.upSet);
                        }
                    }
                }
            }
            catch (SocketTimeoutException st)
            {
                System.out.println("PID: "+this.node.pid+" - Timedout from process: "+this.pid+
                            "\n Ignoring process failure");
                incrementStateResp(Const.UNCERTAIN);
            }
        }
        return 0;
    }
}

class ThreePhaseCommit
{
    public static boolean isRecovering(String pid)
    {
        // Find out if the process is newly created or recovering
        boolean recover=false;
        File f = new File(pid+".log");
        if(f.exists())
            recover=true;
        return recover;
    }

    public static void main(String[] args)
    {
        System.out.println("-----Three Phase Commit Protocol-----");
        // Obtain the id and port of the process
        StringTokenizer idPort = new StringTokenizer(args[0], ":");
        String pid = idPort.nextToken(); 
        // participant or coordinator mode. initially none
        int mode = -1;
        // stage in which coordinator or participant is in (VR/SR)
        String stage = "";
        // creating the node.
        Participant node = new Participant(Integer.parseInt(pid),Integer.parseInt(idPort.nextToken())); 
        // thread executors 
        ExecutorService threadExecutor = null;
        if (mode == -1)
        {
            // Initialization of node
            System.out.println("3PC: Initialization Phase");
            if (args.length > 1)
            {
                //Coordinator node
                // Obtain set of hosts from host port pairs
                System.out.println("PID: "+pid+" - Initializing Coordinator node");
                // initialize a 10 elelment arraylist with port '0'
                ArrayList<Integer> otherHosts = new ArrayList<Integer>();
                for (int j=0;j<Const.MAX_PARTICIPANTS;j++)
                {
                    // initially set all ports to '0' to signify dead hosts
                    otherHosts.add(Const.DEAD_HOST);
                }
                // add coordinator in the host list
                otherHosts.set(Integer.parseInt(pid),node.port);
                // Parse arguments to obtain other nodes
                for (int i=1;i<args.length;i++)
                {
                    StringTokenizer otherIdPort = new StringTokenizer(args[i], ":");
                    String oPid = otherIdPort.nextToken();
                    otherHosts.set(Integer.parseInt(oPid),Integer.parseInt(otherIdPort.nextToken()));
//                    System.out.println("PID: "+pid+" - Participant PID:"
  //                          +otherHosts.get(Integer.parseInt(oPid))+ " added");
                }
                // invoke coordinator mode
                mode = Const.IM_COORD;
                // Add nodes to the upset of coordinator
                node.upSet.addAll(otherHosts);
            }
            else
            {
                // participant mode
                System.out.println("PID: "+pid+" - Initializing Participant");
                mode = Const.IM_PARTICIPANT;
                // make a blank upset
                for (int j=0;j<Const.MAX_PARTICIPANTS;j++)
                {
                    node.upSet.add(0);
                }
            }
            // Coordinator to send VOTE_REQ and participant to wait for it
            stage = Const.VOTE_REQ;
        }
        // keep doing till killed or read end of algorithm
        boolean protoComplete = false;
        while (true)
        {
            try
            {
                if (mode == Const.IM_COORD && !protoComplete)
                {
                    threadExecutor = Executors.newFixedThreadPool(Const.MAX_PARTICIPANTS);
                    System.out.println("PID: "+node.pid+" - Invoking Coordinator node");
                    node.invokeCoordinator(stage,threadExecutor);
                    System.out.println("PID: "+node.pid+" - 3PC Successfully completed");
                }
                else if (mode == Const.IM_PARTICIPANT && !protoComplete)
                {
                    threadExecutor = Executors.newFixedThreadPool(Const.MAX_PARTICIPANTS);
                    // The participant node
                    System.out.println("PID: "+node.pid+" - Invoking Participant node");
                    if (isRecovering(pid) && node.coordinator == -1)
                    {
                        //System.out.println("PID: "+pid+" - Recovering  from crash");
                        node.recover(); 
                    }
                    node.invokeParticipant(stage, threadExecutor);
                    System.out.println("PID: "+node.pid+" - 3PC Successfully completed");
                }
                // protocol has ended
                protoComplete = true;
            }
            catch(ExecutionException re)
            {
                System.out.println("DEBUG: "+node.pid+" - Caught io: " +re.getMessage());
                if ((re.getMessage()).indexOf(Const.UR_ELECTED) != -1)
                {
                    System.out.println("PID: "+node.pid+" Obtained "+Const.UR_ELECTED+" message.");
                    // Coordinator has died and participant is new coordinator
                     stage  = Const.STATE_REQ;
                     mode = Const.IM_COORD; 
                }
                else if((re.getMessage()).indexOf(Const.STATE_REQ) != -1) 
                {
                     stage  = Const.STATE_REQ_STARTED;
                     mode = Const.IM_PARTICIPANT;  
               }
            }
            catch(InterruptedException re)
            {
                System.out.println("PID: "+node.pid+" Caught io: "+re.getMessage());
            }
            catch(SocketTimeoutException st)
            {
//                System.out.println("DEBUG1: "+node.pid+st.getMessage());
                if (node.currentState.equals(Const.START))  
                {
                    System.out.println("PID: "+node.pid+" - Timeout on Coordinator before "
                            +Const.VOTE_REQ+" was obtained. ABORTING");
                    node.writeLog(Const.ABORT,node.upSet);
                    protoComplete = true;
                }
                else
                {
                    System.out.println("PID: "+node.pid+" - Coordinator crashed. "
                            +" Initiating Election Protocol.");
                    node.upSet.set(node.coordinator,Const.DEAD_HOST);
                    mode = node.initiateElection();
                    stage  = Const.STATE_REQ;
                }
            }
            catch(IOException io)
            {
                System.out.println("Caught: "+io.getMessage());
                if (node.currentState.equals(Const.START))  
                {
                    System.out.println("PID: "+node.pid+" - Coordinator crashed before "
                            +Const.VOTE_REQ+" was obtained. ABORTING");
                    node.writeLog(Const.ABORT,node.upSet);
                    protoComplete = true;
                }
                else
                {
                    System.out.println("PID: "+node.pid+" Coordinator crashed. "
                            +" Initiating Election Protocol.");
                    node.upSet.set(node.coordinator,Const.DEAD_HOST);
                    mode = node.initiateElection();
                    stage  = Const.STATE_REQ;
                }
            }
            catch(Exception e)
            {
                System.out.println("Caught: "+e.getMessage());
               // e.printStackTrace();
            }
        }
    }
}
